package com.lhd.app.job;

import com.lhd.app.processor.PetAgeLabelProcessor;
import com.lhd.app.sink.PetAgeLocalFileSink;
import com.lhd.app.bean.PetAgeLabel;
import com.lhd.app.bean.UserBehavior;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;

import java.util.*;

public class PetAgeLabelJob {
    
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        
        // 创建测试数据
        DataStream<UserBehavior> petBehaviorStream = createPetAgeTestData(env);
        
        // 处理宠物年龄标签
        DataStream<PetAgeLabel> ageLabels = petBehaviorStream
            .keyBy(UserBehavior::getUserId)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
            .process(new PetAgeLabelProcessor())
            .name("pet-age-label-processor");
        
        // 输出到本地文件
        ageLabels.addSink(new PetAgeLocalFileSink("data/pet_age_labels.csv"))
            .name("local-file-sink-pet-age");
        
        env.execute("Pet Age Label Generation Job - Local Output");
    }
    
    private static DataStream<UserBehavior> createPetAgeTestData(StreamExecutionEnvironment env) {
        List<UserBehavior> testData = Arrays.asList(
            // 幼犬用户
            createAgeBehavior("user101", "buy", "狗狗", "幼犬粮", "幼犬狗粮", 
                            System.currentTimeMillis() - 86400000L, "幼犬"),
            createAgeBehavior("user101", "cart", "狗狗", "幼犬玩具", "幼犬玩具", 
                            System.currentTimeMillis() - 172800000L, "幼犬"),
            
            // 成年猫用户
            createAgeBehavior("user102", "buy", "猫咪", "成猫粮", "成猫猫粮", 
                            System.currentTimeMillis() - 86400000L, "成年"),
            createAgeBehavior("user102", "favorite", "猫咪", "成猫护理", "成猫保健品", 
                            System.currentTimeMillis() - 172800000L, "成年"),
            
            // 老年狗用户
            createAgeBehavior("user103", "buy", "狗狗", "老年犬粮", "老年狗粮", 
                            System.currentTimeMillis() - 86400000L, "老年"),
            createAgeBehavior("user103", "browse", "狗狗", "老年护理", "老年犬保健品", 
                            System.currentTimeMillis() - 172800000L, "老年")
        );
        
        return env.fromCollection(testData);
    }
    
    private static UserBehavior createAgeBehavior(String userId, String behaviorType,
                                                String cat1, String cat2, String cat3, 
                                                Long timestamp, String ageStage) {
        UserBehavior behavior = new UserBehavior();
        behavior.setUserId(userId);
        behavior.setBehaviorType(behaviorType);
        behavior.setCategory1(cat1);
        behavior.setCategory2(cat2);
        behavior.setCategory3(cat3);
        behavior.setTimestamp(timestamp);
        
        Map<String, String> attributes = new HashMap<>();
        attributes.put("age_stage", ageStage);
        attributes.put("title", cat2 + " " + cat3);
        behavior.setItemAttributes(attributes);
        
        return behavior;
    }
}